feat: Send incremental state as checkpoints after the resources it covers - #2632

🤖 I have created a release *beep* *boop*
---


## [4.99.0](v4.98.0...v4.99.0) (2026-10-19)


### Features

* Send incremental state as checkpoints after the resources it covers ([#2632](#2632)) ([a4bba72](a4bba72))


### Bug Fixes

* **deps:** Update aws-sdk-go-v2 monorepo ([#2629](#2629)) ([fae9da3](fae9da3))
* **deps:** Update aws-sdk-go-v2 monorepo ([#2635](#2635)) ([9f96272](9f96272))
* **deps:** Update module github.com/cloudquery/plugin-pb-go to v1.28.0 ([#2625](#2625)) ([b1a7243](b1a7243))
* **deps:** Update module github.com/cloudquery/plugin-pb-go to v1.28.1 ([#2627](#2627)) ([fc17c1b](fc17c1b))
* **deps:** Update module github.com/cloudquery/plugin-pb-go to v1.28.2 ([#2634](#2634)) ([1e815b0](1e815b0))
* Don't drop work queued by the last active table in shuffle-queue ([#2633](#2633)) ([223018b](223018b))
* Make the JSON type schema match the JSON that the encoder writes ([#2620](#2620)) ([befd5cc](befd5cc))

---
This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please).

#### Summary

Brings the source side of state checkpoints (cloudquery/plugin-pb#42, cloudquery/plugin-sdk#2632, cloudquery/cloudquery#23546) to the Python SDK. When the CLI sets `Sync.Request.supports_checkpoints`, flushed state goes on the sync stream as `Sync.MessageCheckpoint` instead of being written straight to the backend. The CLI then persists it only after every destination commits the records synced before it.

Why we want this: some of our Python syncs run for many hours, often limited by provider rate limits. A transient failure late in the sync currently throws away all the progress, because state is only safe to write once the sync ends. With checkpoints, a failed sync keeps the cursors whose records were committed, and the next run resumes from there.

Opt-in, and unchanged unless both sides opt in:

```python
self._scheduler = Scheduler(concurrency=...)
self._state_client = StateClientBuilder.build(
    backend_options=options.backend_options,
    checkpoints=self._scheduler.checkpoints,  # new, optional
)
self._scheduler.set_post_sync_hook(self._state_client.flush)

# in a table resolver, with one cursor per form
key = state_key("responses", client_id=client.id(), form_id=form["id"])
for item in api.list_responses(form["id"], since=state_client.get_key(key)):
    yield item
    state_client.set_key(key, item["updated_at"])
    state_client.flush()  # a checkpoint, sent once this item and its relations are on the stream
```

Without `checkpoints`, or when the CLI doesn't support them, `flush()` writes to the backend as it does today.

#### Ordering

This follows the Go SDK's semantics:
- A state change made inside a table resolver is tied to that resolver call's scope. The scheduler records the scope in a thread-local while the resolver runs.
- On `flush()`, each group of keys becomes a checkpoint. It goes on the scheduler's queue once every resource those scopes produced before the flush is on it, including the relations those resources spawned.
  - Child resolvers run asynchronously on the depth pools, so the checkpoint waits for every child scope that was still open at flush time.
  - Unlike Go, the Python scheduler has no batcher: resources reach the stream in queue order, so nothing needs flushing first.
- A checkpoint also waits for earlier unsent checkpoints of the same keys. The state table has no `version` column, so the CLI relies on stream order.
- Changes made outside table resolvers wait for nothing. Flush them after the resources they summarize, for example in the post-sync hook. Checkpoints flushed by the post-sync hook are now sent before `Scheduler.sync` returns.
- Flushing in checkpoint mode while no sync is running raises an error instead of dropping the state.

#### Granularity

Two things set how much a checkpointing sync writes, and today plugin authors and users control only one of them:

- **How many keys change between commits.** That's the plugin's choice: one cursor per table, or one per parent, account or other partition. This PR adds `state_key(table, *, client_id=None, **partition)`, so plugins choose the granularity without each inventing a key format. Keys are deterministic, partition fields are sorted, and separators are escaped, so tables, clients and partitions never share a key. Each key is one state row, and a partition's checkpoint waits only for the resolver that set it. Finer keys mean a resumed sync re-fetches less, at the cost of more rows per commit.
- **How often the CLI commits.** The CLI merges checkpoints and commits at most every 30 seconds (`checkpointInterval` in `cli/cmd/sync_v3_checkpoints.go`). Each commit flushes every destination, then writes every key changed since the last commit to the state backend. The interval is hard-coded.

We'd like the commit interval to be configurable per sync, for example in the source spec next to `backend_options`, including an option to commit only at the end. The right value depends on the destination and the sync, not on the plugin:
- **Destination batches:** each flush ends the destination's batch, so a 30-second interval caps batch size by time. For S3 it also writes a new object per table every 30 seconds.
- **State backend load:** a sync with per-partition keys writes up to one row per partition per commit to a relational state backend.
- **Sync length:** for a sync that runs many hours against rate-limited APIs, committing every few minutes loses little on failure and writes far less. A short sync gains nothing from mid-sync commits.

Keeping the plugin-side flush cheap matters too, and it is: `flush()` in checkpoint mode only puts a message on the sync stream, and nothing touches a database until the CLI commits.

#### Other changes

- **State client:** now thread-safe. It clears changed keys after a successful flush (previously every flush rewrote every key changed since the start), and it keeps them after a failed one so a later flush retries them.
- **Backend options:** `BackendOptions` gains `supports_checkpoints`. The server now builds `BackendOptions` from the request instead of passing the protobuf message through.
- **Server guard:** the server refuses to send a checkpoint when the CLI didn't ask for them.
- **Key helper:** `state_key` is exported from `cloudquery.sdk.stateclient`.

#### Depends on

The `plugin-pb-python` bindings with the new messages (cloudquery/plugin-pb-python#250), which aren't released yet (release PR cloudquery/plugin-pb-python#225). `pyproject.toml` requires `>=0.0.55`, but `uv.lock` still needs updating after that release, so CI will fail until then. That's why this is a draft.

#### Open questions

1. **Failed relations.** Like Go, a parent's checkpoint still goes out when one of its child resolvers fails. For a parent cursor such as `updated_since`, that can skip the failed parent's children for good. Would you take an opt-in to hold back a scope's checkpoints when anything in its subtree errored?
2. **`version` column.** Should the Python state table get the `version` column, to match the Go client?
3. **Python destinations.** Python destination plugins don't implement `Write.MessageFlush` yet. I left that for a separate PR, since the CLI only enables checkpoints when every destination supports flushing.
4. **Full syncs.** Checkpoints only help tables that keep cursors. Resuming long full (non-incremental) syncs, and the interaction with `delete-stale`, is a bigger question. I'd like to raise it in a separate issue.

#### Testing

- New tests in `tests/scheduler/checkpoints.py`, `tests/serve/checkpoints.py` and `tests/stateclient/keys.py`. They cover:
  - parent checkpoints following all child resources of the parents they cover, with randomized child latency
  - child checkpoints
  - per-key order under concurrent flushes
  - post-sync hook checkpoints
  - backend writes without checkpoints
  - failed-flush retry
  - the builder and server gating
  - `state_key` ordering, escaping and uniqueness
- Removing either the wait on child scopes or the per-key ordering makes the parent-checkpoint test fail.
- The full suite passes locally against `plugin-pb-python` main, except `test_plugin_package`, which builds a Docker image and fails the same way locally on `main`.